package hasthi.tools;

import static hasthi.common.constants.HasthiConstants.JOIN_MANAGMENT_RING_ACTION;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Map;

import javax.xml.namespace.QName;

import hasthi.common.BaseContext;
import hasthi.common.DistributedLogger;
import hasthi.common.HasthiException;
import hasthi.common.RemoteProcessorException;
import hasthi.common.Utils;
import hasthi.common.XmlBeanUtils;
import hasthi.common.HasthiException.FaultCode;
import hasthi.common.constants.HasthiConstants;
import hasthi.common.constants.HasthiConstants.MngState;
import hasthi.container.CommunicationClient;
import hasthi.container.MessageContext;
import hasthi.container.NodeAddress;
import hasthi.container.SimpleMessageContext;
import hasthi.container.UnicastContianier;
import hasthi.container.p2p.MulticastContianier;
import hasthi.manager.CoordinatorHeartbeatDocument;
import hasthi.manager.CoordinatorInvitaionStatus;
import hasthi.manager.ElectionData;
import hasthi.manager.GetCoordinatorDocument;
import hasthi.manager.GetCoordinatorResponseDocument;
import hasthi.manager.GetLiveManagersDocument;
import hasthi.manager.GetLiveManagersResponseDocument;
import hasthi.manager.JoinManagementRingDocument;
import hasthi.manager.JoinManagementRingResponseDocument;
import hasthi.manager.ManagedResourceType;
import hasthi.manager.ManagerAddress;
import hasthi.manager.ManagerClientUtils;
import hasthi.manager.ManagerContext;
import hasthi.manager.ManagerHeartBeatResponseDocument;
import hasthi.manager.ManagerHeartbeatDocument;
import hasthi.manager.NewCoordinatorInvitationDocument;
import hasthi.manager.NewCoordinatorInvitationResponseDocument;
import hasthi.manager.CoordinatorHeartbeatDocument.CoordinatorHeartbeat;
import hasthi.manager.CoordinatorInvitaionStatus.Enum;
import hasthi.manager.GetCoordinatorResponseDocument.GetCoordinatorResponse;
import hasthi.manager.JoinManagementRingResponseDocument.JoinManagementRingResponse;
import hasthi.manager.ManagerHeartBeatResponseDocument.ManagerHeartBeatResponse;
import hasthi.manager.ManagerHeartbeatDocument.ManagerHeartbeat;
import hasthi.manager.NewCoordinatorInvitationResponseDocument.NewCoordinatorInvitationResponse;
import hasthi.metamodel.ChangeBuffer;
import hasthi.metamodel.ChangeEvent;
import hasthi.metamodel.MixedResourceModel;
import hasthi.metamodel.ResourceUtils;
import hasthi.metamodel.ChangeEvent.ChangeType;
import hasthi.metamodel.resource.ManagedResource;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class ManagementRingTools {
    private static Log log = LogFactory.getLog(HasthiConstants.LOGGER_NAME);

    /**
     * Find the coordinator by performing a anycast on the P2P ring
     * 
     * @param managerContext
     * @param retryLimit -
     *                number of time to retry, -1 mean try till it is sucessful
     * @return
     * @throws HasthiException
     * @throws InterruptedException
     */

    public static GetCoordinatorResponse findCoordinator(ManagerContext managerContext,
            int retryLimit) throws HasthiException {
        // try {
        GetCoordinatorDocument document = GetCoordinatorDocument.Factory.newInstance();
        document.addNewGetCoordinator().setRequesterAddress(managerContext.getMyAddress());

        MulticastContianier p2pContainer = managerContext.getMulticastContianer();

        GetCoordinatorResponse response = null;
        int retryCount = 0;

        // we do anycast
        while (response == null || response.getAddress() == null) {
            MessageContext messageContext = new SimpleMessageContext(document, "AnyCast",
                    HasthiConstants.GET_COORDINATOR_ACTION);
            GetCoordinatorResponseDocument responseDocument = (GetCoordinatorResponseDocument) p2pContainer
                    .anyCastAndReceive(messageContext);
            if (responseDocument != null) {
                response = responseDocument.getGetCoordinatorResponse();
            }
            retryCount++;

            if (retryLimit != -1 && retryCount >= retryLimit) {
                break;
            }
        }
        return response;

        // Environment env = new Environment();
        // env.getParameters().setString("nat_search_policy", "never");
        // NodeIdFactory nidFactory = new RandomNodeIdFactory(env);
        //
        // GetCoordinatorResponseDocument responseDocument;
        // AbstractXmlMesssage responseMessage = null;
        //           
        //
        // // We keep trying till we find a coordinator. We create a Random
        // // NodeID and send a message to that
        // // Node ID effectivley achiving anycast
        //
        // List<String> ids = new ArrayList<String>();
        //            
        //
        // // following code keep sending out few messages and check reponses
        // // till it recieve a address
        // while ((response == null || response.getAddress() == null)) {
        // log.debug(managerContext.getMyAddress().getSoapAddress()
        // + " tyring for get a coordinator");
        // int redundancyCount = 3;
        // // XmlMultiCastMessage message = new
        // // XmlMultiCastMessage(document,p2pContainer.getNode().getId(),
        // // ManagerConstants.GET_COORDINATOR_ACTION);
        // // ids[0] = message.getMessageID();
        // // p2pContainer.anyCast(message);
        // for (int i = 0; i < redundancyCount; i++) {
        // Id randomAddress = nidFactory.generateNodeId();
        // XmlMessage probeMessage = new XmlMessage(document,
        // p2pContainer.getNode()
        // .getId(), ManagerConstants.GET_COORDINATOR_ACTION);
        // ids.add(probeMessage.getMessageID());
        // p2pContainer.sendMessage(randomAddress, probeMessage);
        // }
        //
        // Thread.sleep(1000);
        // // filter ignore the messages with null as address
        // responseMessage = p2pContainer.checkForReplies(ids, new
        // MessageFilter() {
        // public boolean usefulMessage(AbstractXmlMesssage responseMessage) {
        // try {
        // if (responseMessage != null) {
        // GetCoordinatorResponseDocument responseDocument =
        // (GetCoordinatorResponseDocument) responseMessage
        // .getBodyContent();
        // if (responseDocument != null
        // && responseDocument.getGetCoordinatorResponse() != null
        // && responseDocument.getGetCoordinatorResponse()
        // .getAddress() != null) {
        // return true;
        // }
        // }
        // return false;
        // } catch (HasthiException e) {
        // e.printStackTrace();
        // return false;
        // }
        // }
        // });
        // if (responseMessage != null) {
        // responseDocument = (GetCoordinatorResponseDocument) responseMessage
        // .getBodyContent();
        // if (responseDocument != null) {
        // response = responseDocument.getGetCoordinatorResponse();
        // }
        // }
        // if (retryLimit != -1 && retryCount >= retryLimit) {
        // break;
        // } else {
        // redundancyCount++;
        // }
        // }
        // return response;
        // } catch (InterruptedException e) {
        // throw new HasthiException("Interupted While looking for Coordinator",
        // e, FaultCode.LocalError);
        // }
    }

    public static NewCoordinatorInvitationResponseDocument inviteTobeCoordinator(
            ManagerAddress address, ManagerContext managerContext,
            ManagerAddress oldCoordinatorAddress, long oldCoordinatorTimestamp)
            throws RemoteProcessorException, HasthiException {
        NewCoordinatorInvitationDocument invitationDocument = ManagerClientUtils
                .createNewCoordinatiorInvitation(oldCoordinatorAddress, oldCoordinatorTimestamp);
        NewCoordinatorInvitationResponseDocument responseDocument = null;
        MessageContext messageContext = new SimpleMessageContext(invitationDocument, address
                .getSoapAddress(), HasthiConstants.NEWCOORDINATOR_INVITATION_ACTION);
        responseDocument = (NewCoordinatorInvitationResponseDocument) managerContext
                .getUnicastContainer().sendReceive(messageContext);
        return responseDocument;
    }

    public static boolean inviteSucessor(ManagerContext managerContext, ManagerAddress sucessor,
            ManagerAddress oldCoordinator, long coordinatortimestamp)
            throws RemoteProcessorException, HasthiException {
        DistributedLogger dLogger = managerContext.getDlogger();

        NewCoordinatorInvitationDocument invitationDocument = ManagerClientUtils
                .createNewCoordinatiorInvitation(oldCoordinator, coordinatortimestamp);

        NewCoordinatorInvitationResponseDocument responseDocument = null;
        /**
         * try to contact a sucessors and invite them to a election
         */
        if (sucessor == null) {
            ManagerAddress[] sucessors = managerContext.getSucessorList();
            for (int i = 0; i < sucessors.length && responseDocument == null; i++) {
                try {
                    MessageContext messageContext = new SimpleMessageContext(invitationDocument,
                            sucessors[i].getSoapAddress(),
                            HasthiConstants.NEWCOORDINATOR_INVITATION_ACTION);
                    responseDocument = (NewCoordinatorInvitationResponseDocument) managerContext
                            .getUnicastContainer().sendReceive(messageContext);
                    log.info("Inviting " + Utils.getID(sucessors[i]) + " "
                            + Utils.printSucessorList(sucessors));
                } catch (Throwable e) {
                    e.printStackTrace();
                }
            }
        } else {
            MessageContext messageContext = new SimpleMessageContext(invitationDocument, sucessor
                    .getSoapAddress(), HasthiConstants.NEWCOORDINATOR_INVITATION_ACTION);
            responseDocument = (NewCoordinatorInvitationResponseDocument) managerContext
                    .getUnicastContainer().sendReceive(messageContext);
            log.info("Inviting " + Utils.getID(sucessor) + " "
                    + Utils.printSucessorList(managerContext.getSucessorList()));
        }

        if (responseDocument != null) {
            NewCoordinatorInvitationResponse newCoordinatorInvitationResponse = responseDocument
                    .getNewCoordinatorInvitationResponse();
            Enum status = newCoordinatorInvitationResponse.getStatus();
            if (status.equals(CoordinatorInvitaionStatus.COORDINATOR_ALREADY_EXISTS)
                    || status.equals(CoordinatorInvitaionStatus.COORDINATOR_NOT_DOWN)
                    || status.equals(CoordinatorInvitaionStatus.IAM_ALREADY_COORDINATOR)) {
                /**
                 * Which means Sucessor says there is a coordinator, which is
                 * very unlikely. We will give it a chance but send a wranning
                 */
                ManagerAddress newCoordinatorAddress = newCoordinatorInvitationResponse
                        .getCoordinatorAddress();
                long coordinatorTimestamp = newCoordinatorInvitationResponse
                        .getCoordinatorTimestamp();
                managerContext.setCoordinatorAddress(newCoordinatorAddress, coordinatorTimestamp);
                if (status.equals(CoordinatorInvitaionStatus.COORDINATOR_ALREADY_EXISTS)) {
                    dLogger.warrning("Sucessor claimed ("
                            + newCoordinatorInvitationResponse.getCoordinatorAddress()
                                    .getSoapAddress() + ") is coordinator when he was invited.");
                } else if (status.equals(CoordinatorInvitaionStatus.COORDINATOR_NOT_DOWN)) {
                    dLogger.warrning("We belive coordinator("
                            + newCoordinatorAddress.getSoapAddress()
                            + ") is down, but sucessor claims it is up");
                }
            } else if (status.equals(CoordinatorInvitaionStatus.ELECTION_ALREADY_STARTED)
                    || status.equals(CoordinatorInvitaionStatus.ELECTION_STARTED)) {
                // Sucessor has start a election We wait for sucessor to take
                // over
            } else {
                // means there is a better sucessor up. We will let this go
                // hoping somebody else
                // might start a election. If that does not happen cooridnator
                // will be started by probabiliastic
                // model
            }
            return true;
        } else {
            return false;
        }
    }

    public static String findCoordinatorFromBootstrapNode(UnicastContianier container,
            BaseContext baseContext) throws InterruptedException, HasthiException {
        try {
            String bootstrapSoapAddress = "http://" + baseContext.getBoostrapHost() + ":"
                    + baseContext.getBoostrapSoapPort();
            GetLiveManagersDocument document = GetLiveManagersDocument.Factory.newInstance();
            document.addNewGetLiveManagers();
            String coordinationUrl = null;
            // Following code try to get Managers from the bootstrap node and
            // try to
            // connect to the that
            // Manager and find the coordinator address. Once the coordinator is
            // found
            int retryCount = 0;
            while (coordinationUrl == null && retryCount < 5) {
                // Contact Boostrap node
                MessageContext message1 = new SimpleMessageContext(document, bootstrapSoapAddress,
                        HasthiConstants.GET_LIVEMANEGERS_ACTION);
                GetLiveManagersResponseDocument responseDocument = (GetLiveManagersResponseDocument) container
                        .sendReceive(message1);
                ManagerAddress[] addresses = responseDocument.getGetLiveManagersResponse()
                        .getManagerAddressArray();
                // contact each Manager
                if (addresses != null) {
                    for (int i = 0; i < addresses.length && coordinationUrl == null; i++) {
                        try {
                            GetCoordinatorDocument coordinatorDocument = GetCoordinatorDocument.Factory
                                    .newInstance();
                            coordinatorDocument.addNewGetCoordinator();
                            MessageContext message2 = new SimpleMessageContext(coordinatorDocument,
                                    addresses[i].getSoapAddress(),
                                    HasthiConstants.GET_COORDINATOR_ACTION);
                            GetCoordinatorResponseDocument coordinatorResponse = (GetCoordinatorResponseDocument) container
                                    .sendReceive(message2);
                            ManagerAddress managerAddress = coordinatorResponse
                                    .getGetCoordinatorResponse().getAddress();
                            if (managerAddress != null) {
                                coordinationUrl = managerAddress.getSoapAddress();
                            }
                        } catch (Throwable e) {
                            // If there is error, we ignore and try next address
                            // till we find a coordinator
                            e.printStackTrace();
                        }
                    }
                }
                // If no coodinator found, may be election is going on, Wait
                // TODO print logging message
                if (coordinationUrl == null) {
                    Thread.sleep(60000);
                    retryCount++;
                }
            }
            return coordinationUrl;
        } catch (RemoteProcessorException e) {
            throw new HasthiException("Error coonecting to bootstrap node", e, e.getFaultCode());
        }
    }

    public static void joinManagmentRing(ManagerContext managerContext, String coordinatorAddress)
            throws HasthiException {
        UnicastContianier soapContainer = managerContext.getUnicastContainer();

        JoinManagementRingDocument joinManagementRingDocument = JoinManagementRingDocument.Factory
                .newInstance();
        joinManagementRingDocument.addNewJoinManagementRing().setAddress(
                managerContext.getMyAddress());
        MessageContext messageContext = new SimpleMessageContext(joinManagementRingDocument,
                coordinatorAddress, JOIN_MANAGMENT_RING_ACTION);
        JoinManagementRingResponseDocument joinManagementRingResponseDocument = (JoinManagementRingResponseDocument) soapContainer
                .sendReceive(messageContext);
        JoinManagementRingResponse joinManagementRingResponse = joinManagementRingResponseDocument
                .getJoinManagementRingResponse();
        managerContext.setMyRank(joinManagementRingResponse.getAssignedRank());
        ManagerAddress[] sucessors = joinManagementRingResponse.getSucessorArray();
        if (sucessors != null) {
            managerContext.setSucessorList(sucessors);
        }
        managerContext.setManagementRingSize(joinManagementRingResponse.getManagerCount());

        // We only set the coordinator after we join the Management ring
        managerContext.setCoordinatorAddress(joinManagementRingResponse.getCoordinatorAddress(),
                joinManagementRingResponse.getCoordinatorTimestamp());
        managerContext.setJoinedCoordinator();
    }

    public static void sendCoordinatorHeartbeat(ManagerContext managerContext)
            throws HasthiException {
        try {
            CoordinatorHeartbeatDocument coordinatorHeartbeatDocument = CoordinatorHeartbeatDocument.Factory
                    .newInstance();
            CoordinatorHeartbeat coordinatorHeartbeat = coordinatorHeartbeatDocument
                    .addNewCoordinatorHeartbeat();
            coordinatorHeartbeat.setAddress(managerContext.getMyAddress());
            coordinatorHeartbeat.setCoordinatorTimestamp(managerContext.getCoordinatorTimeStamp());
            MulticastContianier multicastContianier = managerContext.getMulticastContianer();
            multicastContianier.mulitiCastAndWait(coordinatorHeartbeatDocument,
                    HasthiConstants.COORDINATOR_HEARTBEAT_ACTION, 0, 0);
        } catch (InterruptedException e) {
            throw new HasthiException(e, FaultCode.LocalError);
        }
    }

    public static void sendHeartbeatsAndPorcessPiggyback(ManagerContext managerContext,
            String coordinatorSoapAddress) throws HasthiException {
        MixedResourceModel resourceModel = managerContext.getResourceModel();
        ManagerHeartbeatDocument document;
        /**
         * If we have already joined a ring coordinator, we will only send the
         * changes done. Otherwise we send a summery of all resources hold in
         * this manager
         */
        boolean joinedCoordinator = (managerContext.getManagerState() == MngState.HasCoordinatorJoined);
        if (joinedCoordinator) {
            document = createHeartbeatWithChange(managerContext, resourceModel);
        } else {
            document = createHeartbeatWithAllData(managerContext, resourceModel);
        }

        NodeAddress coordinatorAddress = new NodeAddress(coordinatorSoapAddress);
        CommunicationClient client = managerContext.getUnicastContainer()
                .createCommunicationClient();
        MessageContext message = new SimpleMessageContext(document, coordinatorAddress
                .getAddressAsString(), HasthiConstants.MANAGER_HEARTBEAT_ACTION);
        ManagerHeartBeatResponseDocument responseDocument = (ManagerHeartBeatResponseDocument) client
                .sendReceive(message);

        ManagerHeartBeatResponse managerHeartBeatResponse = responseDocument
                .getManagerHeartBeatResponse();
        ManagerAddress[] sucessors = managerHeartBeatResponse.getSucessorArray();
        managerContext.setSucessorList(sucessors);
        managerContext.setManagementRingSize(managerHeartBeatResponse.getManagerCount());

        if (managerHeartBeatResponse.getResourceNameToRemoveArray() != null) {
            for (QName resourceName : managerHeartBeatResponse.getResourceNameToRemoveArray()) {
                // if you change here you should change ManagerMessage processor
                // as well
                ManagedResource resource = managerContext.getResourceModel().removeResource(
                        resourceName);
                if(resource != null){
                    managerContext.getRuleEngine().removeObject(resource);
                    log.info("[" + managerContext + "]stop managing " + resourceName);
                }

            }
        }
        // If I am assigned a new rank, update it
        if (!joinedCoordinator
                && (managerContext.getManagerState() == MngState.HasCoordinatorNotJoined)) {
            managerContext.setJoinedCoordinator();
        }

        if (managerHeartBeatResponse.getAddedAsNew()) {
            managerContext.setMyRank(managerHeartBeatResponse.getAssignedRank());
            managerContext.setJoinedCoordinator();
        }

        if (!managerContext.amICoordinator()
                && !joinedCoordinator != managerHeartBeatResponse.getAddedAsNew()) {
            log.fatal("[" + managerContext + "]" + (joinedCoordinator ? "" : "Not ")
                    + "Joined coordintor, but "
                    + (managerHeartBeatResponse.getAddedAsNew() ? "" : "Not ")
                    + "added as new via heartbeat");
        }
    }

    public static ManagerHeartbeatDocument createHeartbeatWithChange(ManagerContext managerContext,
            MixedResourceModel resourceModel) throws HasthiException {
        Map<QName, ManagedResourceType> resourceDiffMap = managerContext.getResourceDiffMap();
        ManagerHeartbeatDocument document;
        document = ManagerHeartbeatDocument.Factory.newInstance();
        ManagerHeartbeat heartbeat = document.addNewManagerHeartbeat();
        heartbeat.setAddress(managerContext.getMyAddress());
        /**
         * We build the heartbeat message using the changes. May be we are
         * sending too many changes now. We can only send summary if needed
         */
        synchronized (resourceDiffMap) {
            int changeCount = 0;
            ChangeBuffer changes = resourceModel.getChangeBuffer();
            List<ChangeEvent> changeEvents = changes.reset();
            for (ChangeEvent change : changeEvents) {
                if (ChangeType.Update.equals(change.getChangeType())) {
                    String propertyName = change.getPropertyName().getLocalPart();
                    if (ResourceUtils.isIncludedInSummerized(propertyName)) {
                        ManagedResourceType resourceType = resourceDiffMap.get(change
                                .getResourceID());
                        if (resourceType == null) {
                            resourceType = ManagedResourceType.Factory.newInstance();
                            resourceDiffMap.put(change.getResourceID(), resourceType);
                        }
                        XmlBeanUtils.addNameValuePair(resourceType, change.getPropertyName(),
                                change.getNewValue());
                        changeCount++;
                    }
                }
                // TODO handle other changes types, add/remove. Add is
                // automatically handled. We need to handle removes
            }
            log.debug("[" + managerContext + "]Sending change for " + changeCount + "/"
                    + resourceModel.getResources().size());
            // Xmlbeans complains if we add properties and children to a element
            // mixed. So we set the properties at the end
            for (QName name : resourceDiffMap.keySet()) {
                ManagedResourceType resourceType = resourceDiffMap.get(name);
                resourceType.setName(name);
                resourceType.setChangesOnly(true);
            }
        }
        heartbeat.setPiggybackUpdateArray(resourceDiffMap.values().toArray(
                new ManagedResourceType[0]));
        resourceDiffMap.clear();
        return document;
    }

    public static ManagerHeartbeatDocument createHeartbeatWithAllData(
            ManagerContext managerContext, MixedResourceModel resourceModel) throws HasthiException {
        ManagerHeartbeatDocument document;
        document = ManagerHeartbeatDocument.Factory.newInstance();
        ManagerHeartbeat heartbeat = document.addNewManagerHeartbeat();
        heartbeat.setAddress(managerContext.getMyAddress());
        heartbeat.setAllResourceIncluded(true);

        Collection<QName> resourceNames = resourceModel.getResources();
        List<ManagedResourceType> resources = new ArrayList<ManagedResourceType>(resourceNames.size());
        for (QName resourceName : resourceNames) {
            resources.add(resourceModel.getResource(resourceName).toXml(true));
        }
        heartbeat.setPiggybackUpdateArray(resources.toArray(new ManagedResourceType[0]));
        return document;
    }
    
    public static void joinUsingHeartbeat(ManagerContext managerContext,
            String coordinatorSoapAddress) throws HasthiException {
        MixedResourceModel resourceModel = managerContext.getResourceModel();
        ManagerHeartbeatDocument document;
        document = createHeartbeatWithAllData(managerContext, resourceModel);

        NodeAddress coordinatorNodeAddress = new NodeAddress(coordinatorSoapAddress);
        CommunicationClient client = managerContext.getUnicastContainer()
                .createCommunicationClient();
        MessageContext message = new SimpleMessageContext(document, coordinatorNodeAddress
                .getAddressAsString(), HasthiConstants.MANAGER_HEARTBEAT_ACTION);
        ManagerHeartBeatResponseDocument responseDocument = (ManagerHeartBeatResponseDocument) client
                .sendReceive(message);

        ManagerHeartBeatResponse managerHeartBeatResponse = responseDocument.getManagerHeartBeatResponse();
        ManagerAddress[] sucessors = managerHeartBeatResponse.getSucessorArray();
        managerContext.setSucessorList(sucessors);
        managerContext.setManagementRingSize(managerHeartBeatResponse.getManagerCount());
        managerContext.setCoordinatorAddress(managerHeartBeatResponse.getCoordinatorAddress(), 
                managerHeartBeatResponse.getCoordinatorTimestamp());
        managerContext.setJoinedCoordinator();
        if (managerHeartBeatResponse.getResourceNameToRemoveArray() != null) {
            for (QName resourceName : managerHeartBeatResponse.getResourceNameToRemoveArray()) {
                // if you change here you should change ManagerMessage processor
                // as well
                ManagedResource resource = managerContext.getResourceModel().removeResource(
                        resourceName);
                if(resource != null){
                    managerContext.getRuleEngine().removeObject(resource);
                    log.info("[" + managerContext + "]stop managing " + resourceName);
                }
            }
        }
        if (managerHeartBeatResponse.getAddedAsNew()) {
            managerContext.setMyRank(managerHeartBeatResponse.getAssignedRank());
        }else if(!managerContext.amICoordinator()){
            log.fatal("Join via heartbeat, but not added");
        }
    }
}
